package com.mushuilingfeng.ddd.framework.event;


import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author
 * @date 2024/6/18
 */
public class DefaultDomainEventPublisherSubscribe implements IDomainEventPublisher, IDomainEventSubscribe {
    // 队列
    private final Map<Class<? extends IDomainEvent>, ArrayBlockingQueue<IDomainEvent>> eventQueueMap = new ConcurrentHashMap<>();
    private final Map<Class<? extends IDomainEvent>, ArrayBlockingQueue<IDomainEventListener<? extends IDomainEvent>>> listenerQueueMap = new ConcurrentHashMap<>();
    private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);
    private static final int QUEUE_SIZE = 10000;

    private ArrayBlockingQueue<IDomainEvent> getEventQueue(Class<? extends IDomainEvent> eventCls) {
        return eventQueueMap.computeIfAbsent(eventCls, (cls) -> new ArrayBlockingQueue<>(QUEUE_SIZE));
    }

    @Override
    public void publish(IDomainEvent event) {
        ArrayBlockingQueue<IDomainEvent> queue = getEventQueue(event.getClass());
        queue.add(event);
    }

    @Override
    public void subscribe(Class<? extends IDomainEvent> eventCls, IDomainEventListener<? extends IDomainEvent> listener) {
        ArrayBlockingQueue<IDomainEventListener<? extends IDomainEvent>> queue = listenerQueueMap.get(eventCls);
        if (null == queue) {
            synchronized (listenerQueueMap) {
                queue = listenerQueueMap.get(eventCls);
                if (null == queue) {
                    queue = new ArrayBlockingQueue<>(QUEUE_SIZE);
                    listenerQueueMap.put(eventCls, queue);
                    // 处理
                    scheduledThreadPoolExecutor.scheduleAtFixedRate(new EventConsumerTask(getEventQueue(eventCls), queue), 1000L, 100L,
                            TimeUnit.MILLISECONDS);
                }
            }
        }
        queue.add(listener);
    }


    private static class EventConsumerTask implements Runnable {
        private final ArrayBlockingQueue<IDomainEvent> eventQueue;
        private final ArrayBlockingQueue<IDomainEventListener<? extends IDomainEvent>> listenerQueue;

        public EventConsumerTask(ArrayBlockingQueue<IDomainEvent> eventQueue, ArrayBlockingQueue<IDomainEventListener<? extends IDomainEvent>> listenerQueue) {
            this.eventQueue = eventQueue;
            this.listenerQueue = listenerQueue;
        }

        @SuppressWarnings("all")
        @Override
        public void run() {
            try {
                IDomainEvent event = eventQueue.poll(1, TimeUnit.MILLISECONDS);
                if (null == event) {
                    return;
                }
                for (IDomainEventListener listener : listenerQueue) {
                    try {
                        listener.onEvent(event);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
